package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Code35PageRank2 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("LogDataCompute")
    val sc = new SparkContext(conf)

    val pageReleaseRDD: RDD[(String, List[String])] = sc
      .textFile("spark_code/data/page.txt")
      .map {
        case line =>
          val splitRes: Array[String] = line.split("->")
          val page: String = splitRes(0)
          val outPageList = splitRes(1).split(",").toList
          (page, outPageList)
      }

    var lastPrRes: RDD[(String, List[String], Double)] = pageReleaseRDD
      // 1.整理数据 对 A->B,D 按照 -> 切分页面，同时对指向的多个页面转换成数组或List
      .map {
        case (page, outPageList) =>
          (page, outPageList, 1.0)
      }
    var flagDouble = 1.0
    var cnt:Int = 0

    while (flagDouble > 0.0001) {

      val currentPRRes: RDD[(String, (Double, List[String]))] = lastPrRes
        // 2.第一轮计算 对页面进行打分
        .flatMap {
          case (page, outPageList, pr) => {
            val avgPr = pr / outPageList.size.toDouble
            outPageList.map {
              case page => (page, avgPr)
            }
          }
        }
        // 3.对每个页面统计获得的分数
        .groupBy(_._1)
        .mapValues(_.map(_._2).sum)
        .join(pageReleaseRDD)

      flagDouble = lastPrRes.map {
        case (page, outPageList, pr) => (page, (outPageList, pr))
      }
        .join(currentPRRes)
        .map {
          case (page, ((lastOutPageList, lastPr), (pr, outPageList))) => {
            math.abs(lastPr - pr)
          }
        }
        .sum()
      cnt += 1
      println(s"当前任务执行了${cnt}次,其迭代的阈值为:$flagDouble")

      lastPrRes = currentPRRes.map {
        case (page, (pr, outPageList)) => {
          (page, outPageList, pr)
        }
      }

    }

    lastPrRes
      .foreach(println)

    while(true){}
  }
}
